package org.example.day20250105.flink.jdbc;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.day20250105.ExcelDataHandle;
import org.example.day20250105.domain.OdsXwCycleData;
import org.example.day20250105.flink.DorisConstant;

import java.sql.PreparedStatement;
import java.sql.SQLException;

public class OdsXwCycleDataWriteByFlink {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.fromData(ExcelDataHandle.readCycleData())
                .addSink(JdbcSink.sink(
                        "INSERT INTO `test_db_sl`.`ods_xw_cycle_data` (`cycle_id`, `start_absolute_time`, `end_absolute_time`, `charge_capacity`, `discharge_capacity`," +
                                " `charge_specific_capacity`, `discharge_specific_capacity`, `charge_discharge_efficiency`, " +
                                "`charge_energy`, `discharge_energy`, `charge_specific_energy`, `discharge_specific_energy`," +
                                " `constant_current_charging_ratio`, `constant_current_charging_capacity`, `platform_capacity`, " +
                                "`platform_specific_capacity`, `platform_capacity_efficiency`, `platform_time`, `charge_capacitor`," +
                                " `discharge_capacitor`, `median_voltage`, `capacity_retention_rate`, `charge_millisecond`, " +
                                "`discharge_millisecond`, `net_discharge_capacity`, `net_discharge_energy`, `energy_efficiency`," +
                                " `t1_start_temperature`, `t1_end_temperature`, `t1_max_temperature`, `t1_min_temperature`, " +
                                "`v1_start_voltage`, `v1_end_voltage`, `v1_max_voltage`, `v1_min_voltage`, `t1_charge_max_temperature`," +
                                " `t1_charge_min_temperature`, `t1_discharge_max_temperature`, `t1_discharge_min_temperature`, `file_name`) " +
                                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
                        new JdbcStatementBuilder<OdsXwCycleData>() {
                            @Override
                            public void accept(PreparedStatement ps, OdsXwCycleData data) throws SQLException {
                                ps.setString(1, data.getCycle_id());
                                ps.setString(2, data.getStart_absolute_time());
                                ps.setString(3, data.getEnd_absolute_time());
                                ps.setString(4, data.getCharge_capacity());
                                ps.setString(5, data.getDischarge_capacity());
                                ps.setString(6, data.getCharge_specific_capacity());
                                ps.setString(7, data.getDischarge_specific_capacity());
                                ps.setString(8, data.getCharge_discharge_efficiency());
                                ps.setString(9, data.getCharge_energy());
                                ps.setString(10, data.getDischarge_energy());
                                ps.setString(11, data.getCharge_specific_energy());
                                ps.setString(12, data.getDischarge_specific_energy());
                                ps.setString(13, data.getConstant_current_charging_ratio());
                                ps.setString(14, data.getConstant_current_charging_capacity());
                                ps.setString(15, data.getPlatform_capacity());
                                ps.setString(16, data.getPlatform_specific_capacity());
                                ps.setString(17, data.getPlatform_capacity_efficiency());
                                ps.setString(18, data.getPlatform_time());
                                ps.setString(19, data.getCharge_capacitor());
                                ps.setString(20, data.getDischarge_capacitor());
                                ps.setString(21, data.getMedian_voltage());
                                ps.setString(22, data.getCapacity_retention_rate());
                                ps.setString(23, data.getCharge_millisecond());
                                ps.setString(24, data.getDischarge_millisecond());
                                ps.setString(25, data.getNet_discharge_capacity());
                                ps.setString(26, data.getNet_discharge_energy());
                                ps.setString(27, data.getEnergy_efficiency());
                                ps.setString(28, data.getT1_start_temperature());
                                ps.setString(29, data.getT1_end_temperature());
                                ps.setString(30, data.getT1_max_temperature());
                                ps.setString(31, data.getT1_min_temperature());
                                ps.setString(32, data.getV1_start_voltage());
                                ps.setString(33, data.getV1_end_voltage());
                                ps.setString(34, data.getV1_max_voltage());
                                ps.setString(35, data.getV1_min_voltage());
                                ps.setString(36, data.getT1_charge_max_temperature());
                                ps.setString(37, data.getT1_charge_min_temperature());
                                ps.setString(38, data.getT1_discharge_max_temperature());
                                ps.setString(39, data.getT1_discharge_min_temperature());
                                ps.setString(40, data.getFile_name());
                            }
                        },
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUsername(DorisConstant.USERNAME)
                                .withPassword(DorisConstant.PASSWORD)
                                .withDriverName(DorisConstant.DRIVER_CLASS_NAME)
                                .withUrl(DorisConstant.DORIS_JDBC_URL)
                                .build()
                ));
        env.execute();
    }
}
